package net.wicp.tams.common.binlog.self.reader;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import org.apache.commons.lang3.ArrayUtils;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Result;
import net.wicp.tams.common.apiext.CollectionUtil;
import net.wicp.tams.common.apiext.IOUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.binlog.self.IBinlogRead;
import net.wicp.tams.common.binlog.self.IEventRead;
import net.wicp.tams.common.binlog.self.bean.CountNum;
import net.wicp.tams.common.binlog.self.bean.EventHeader;
import net.wicp.tams.common.binlog.self.bean.Host;
import net.wicp.tams.common.binlog.self.bean.PacketBean;
import net.wicp.tams.common.binlog.self.bean.Pos;
import net.wicp.tams.common.binlog.self.bean.SendMoniter;
import net.wicp.tams.common.binlog.self.constant.BinLogVersion;
import net.wicp.tams.common.binlog.self.constant.Checksum;
import net.wicp.tams.common.binlog.self.constant.EventType;
import net.wicp.tams.common.binlog.self.event.GtidEvent;
import net.wicp.tams.common.binlog.self.event.RotateEvent;
import net.wicp.tams.common.binlog.self.event.rows.RowsEvent;
import net.wicp.tams.common.binlog.self.replication.down.EOFPacket;
import net.wicp.tams.common.binlog.self.replication.down.ErrorPacket;
import net.wicp.tams.common.binlog.self.replication.down.OKPacket;
import net.wicp.tams.common.binlog.self.sender.ISender;
import net.wicp.tams.common.thread.threadlocal.PerthreadManager;

@Slf4j
public class BinglogReadDump extends BufferedInputStream implements IBinlogRead {
	private String fileName;
	private boolean curFile = false;// 是否只处理连接时指定的文件，默认为否，true：是，适用于补日志
	// protected long eventNum = 0;// 已解析成功的event数量
	private final ISender[] senders;
	private long curPos;
	private long curTime;
	private long masterServerId;
	private String gtids;
	private final String uuid;// 主服务器source
	private final Checksum checksum;
	private String slaveGtids = "";// 从服务器第一个字符为","
	private long parserDelaytime = 0;// 开始解析的延时时间
	private long parserAndSendTime = 0;// 解析用时

	public BinglogReadDump(String uuid, Checksum checksum, InputStream in, ISender... senders) {
		super(in);
		this.uuid = uuid;
		this.checksum = checksum;
		if (senders != null && senders.length > 0) {// senders.length > 0不能
			this.senders = senders;
		} else {
			this.senders = null;
		}
	}

	public BinglogReadDump(String uuid, Checksum checksum, InputStream in, String dbPattern, String tbPattern,
			ISender... senders) {
		this(uuid, checksum, in, senders);
		Host host = (Host) PerthreadManager.getInstance().createValue("zorro-host",Host.class).get(new Host());
		host.setDbPattern(dbPattern);
		host.setTbPattern(tbPattern);
	}

	/***
	 * 不用检查魔数，因为dump不支持小于4的pos
	 */
	@Override
	public Result checkHead() {
		try {
			PacketBean bin = PacketBean.buildEventBean(in, fileName, gtids, checksum);
			if (bin.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
				final ErrorPacket binPacket = new ErrorPacket(bin);
				log.error("读事件包出错：[{}]", binPacket.getBody().toJSONString());
				return Result.getError(String.format("读事件包出错：[%s]", binPacket.getBody().toJSONString()));
			}
			BinLogVersion binLogVersion = getVersion(bin.getEventBean().getHead());
			if (binLogVersion == null || binLogVersion != BinLogVersion.v4) {
				return Result.getError(String.format("只支持v4的binlog，不支持的binlog版本:[%s]",
						binLogVersion == null ? "" : binLogVersion.name()));
			}
			bin.getEventBean().getEventRead().parseBody();
		} catch (IOException e) {// 失败后用文件方式设置
			Properties props = IOUtil.fileToProperties("/conf/binlog/FormatDescription.properties",
					BinglogReadDump.class);
			JSONObject obj = new JSONObject();
			for (Object key : props.keySet()) {
				obj.put(String.valueOf(key), Integer.parseInt(String.valueOf(props.get(key))));
			}
			EventType.setFormatDescription(obj);
		}

		return Result.getSuc();
	}

	@Override
	public void read(long pos) throws IOException {
		Host host = (Host) PerthreadManager.getInstance().createValue("zorro-host",Host.class).get();
		if (host == null || !EventType.isInit(host)) {
			Result checkRet = checkHead();
			if (!checkRet.isSuc()) {
				throw new RuntimeException(checkRet.getMessage());
			}
		}
		while (true) {
			// long time = System.currentTimeMillis();
			PacketBean bin = PacketBean.buildEventBean(in, fileName, gtids, checksum);
			// LogBackUtil.logDebugTime(log, "组装包", time);
			curPos = bin.getEventBean().getBeginHead();
			curTime = bin.getEventBean().getHead().getTimestamp();
			if (bin.getPacketMarker() == ErrorPacket.PACKET_MARKER) {
				final ErrorPacket binPacket = new ErrorPacket(bin);
				log.error("读事件包出错：[{}]", binPacket.getBody().toJSONString());
				break;
			} else if (bin.getPacketMarker() == EOFPacket.PACKET_MARKER) {
				final EOFPacket binPacket = new EOFPacket(bin);
				log.error("读事件包结束：[{}]", binPacket.getBody().toJSONString());
			} else if (bin.getPacketMarker() == OKPacket.PACKET_MARKER) {
				if (bin.getEventBean() == null) {
					break;
				}
				long parserDelaytimeTemp = System.currentTimeMillis() / 1000
						- bin.getEventBean().getHead().getTimestamp();// 开始解析的延时时间

				IEventRead reader = bin.getEventBean().getEventRead();
				if (bin.getEventBean().getHead().getEventType() == EventType.STOP_EVENT) {
					reader.parseBody();
					if (curFile) {
						break;
					}
				} else if (bin.getEventBean().getHead().getEventType() == EventType.ROTATE_EVENT) {// 跳转事件
					RotateEvent rotateEvent = (RotateEvent) bin.getEventBean().getEventRead();
					rotateEvent.parseBody();
					fileName = rotateEvent.getNextBinlogName();
					log.info("跳转到下一个文件:【{}】，位置:【{}】", fileName, rotateEvent.getNextPosition());
					continue;
				} else if (bin.getEventBean().getHead().getEventType() == EventType.GTID_EVENT) {// TOdO
																									// this.canGtid=false处理
																									// gtid事件
					GtidEvent gtidEvent = (GtidEvent) bin.getEventBean().getEventRead();
					gtidEvent.parseBody(senders);
					if (StringUtil.isNotNull(uuid) && uuid.equals(gtidEvent.getSource())) {// 当做主备时会变化source
						if (StringUtil.isNull(slaveGtids)) {
							gtids = gtidEvent.getGtid();
						} else {
							gtids = String.format("%s,%s", gtidEvent.getGtid(), slaveGtids);
						}
					} else {
						log.info("------------------做主备切换,原主机源[{}],切换源[{}}]--------------------------------", uuid,
								gtidEvent.getSource());
					}
					continue;
				} else if (bin.getEventBean().getHead().getEventType() == EventType.TABLE_MAP_EVENT
						|| bin.getEventBean().getHead().getEventType() == EventType.FORMAT_DESCRIPTION_EVENT) {
					reader.parseBody();
				} else if (reader instanceof RowsEvent) {
					long beginParseTimeTemp = System.currentTimeMillis();
					Result ret = reader.parseBody(senders);
					// LogBackUtil.logDebugTime(log, "解析并发送", time);
					// senders
					if (ret != null) {
						if (!ret.isSuc() && StringUtil.isNotNull(ret.getMessage())) {
							log.error("errormsg:[{}]", ret.getMessage());
						} else {
							parserDelaytime += parserDelaytimeTemp;// mysql到zorro的延时时间
							parserAndSendTime += (System.currentTimeMillis() - beginParseTimeTemp);// 解析和发送用时
						}
					}

				} else {
					log.info("暂不支持的事件:[{}]", bin.getEventBean().getHead().getEventType());
				}

			} else {
				log.error("不认识的包,包识别码[{}]", bin.getPacketMarker());
			}
		}
	}

	// 需要在第9个位开始读,读13位
	protected BinLogVersion getVersion(EventHeader firstHead) {
		if (firstHead == null) {
			return null;
		}
		BinLogVersion retobj = null;
		switch (firstHead.getEventType()) {
		case FORMAT_DESCRIPTION_EVENT:
			retobj = BinLogVersion.v4;
			break;
		case START_EVENT_V3:
			long eventSize = firstHead.getEventSize();
			if (eventSize == (13 + 56)) {
				return BinLogVersion.v1;
			} else if (eventSize == (19 + 56)) {
				return BinLogVersion.v3;
			}
			break;
		default:
			break;
		}
		return retobj;
	}

	//////////////////////////////////////////////////// 下面是一些扩展方法////////////////////////////////////////////////////////////////////////////
	/**
	 * 得到当前位置
	 * 
	 * @return
	 */
	public long getCurPos() {
		return this.pos;
	}

	public String getFileName() {
		return fileName;
	}

	public void setFileName(String fileName) {
		this.fileName = fileName;
	}

	public boolean isCurFile() {
		return curFile;
	}

	public void setCurFile(boolean curFile) {
		this.curFile = curFile;
	}

	@Override
	public Pos curPos() {
		Pos ret = new Pos();
		ret.setFileName(fileName);
		ret.setPos(curPos);
		ret.setTime(curTime);
		ret.setMasterServerId(masterServerId);
		ret.setGtids(gtids);
		return ret;
	}

	@Override
	public void resetNum() {
		// this.insertNum = 0;
		// this.updateNum = 0;
		// this.deleteNum = 0;
		this.parserDelaytime = 0;
		this.parserAndSendTime = 0;
		SendMoniter initbean = SendMoniter.builder().insertNum(0).updateNum(0).deleteNum(0).sendTimes(0).build();
		for (ISender sender : this.senders) {
			sender.initMoniter(initbean);
		}
	}

	@Override
	public CountNum getCountNum() {
		long sendtimes = 0;
		for (ISender iSender : this.senders) {
			sendtimes += iSender.getMoniter().getSendTimes();
		}
		SendMoniter sendMoniter = this.senders[0].getMoniter();
		CountNum countNum = CountNum.builder().insertNum(sendMoniter.getInsertNum())
				.updateNum(sendMoniter.getUpdateNum()).deleteNum(sendMoniter.getDeleteNum())
				.parserDelayTime(parserDelaytime).parserAndSendTime(parserAndSendTime).sendTime(sendtimes).build();
		return countNum;
	}

	public void initCountNum(CountNum countNum) {
		if (countNum == null) {
			return;
		}
		// this.insertNum = countNum.getInsertNum();
		// this.updateNum = countNum.getUpdateNum();
		// this.deleteNum = countNum.getDeleteNum();
		this.parserDelaytime = countNum.getParserDelayTime();
		this.parserAndSendTime = countNum.getParserAndSendTime();
		long sendtime = countNum.getSendTime() / this.senders.length;// 设置平均时间
		SendMoniter intobj = SendMoniter.builder().insertNum((int) countNum.getInsertNum())
				.updateNum((int) countNum.getUpdateNum()).deleteNum((int) countNum.getDeleteNum()).sendTimes(sendtime)
				.build();
		for (ISender send : this.senders) {
			send.initMoniter(intobj);
		}
	}

	public void setMasterServerId(long masterServerId) {
		this.masterServerId = masterServerId;
	}

	public void setGtids(String gtids) {
		if (StringUtil.isNull(gtids)) {
			this.gtids = null;
			this.slaveGtids = null;
			// this.canGtid = false;
		} else {
			// this.canGtid = true;
			this.gtids = gtids.replace("\n", "");
			String[] gtidsAry = this.gtids.split(",");

			for (String gtid : gtidsAry) {
				if (gtid.startsWith(uuid)) {
					gtidsAry = (String[]) ArrayUtils.removeElement(gtidsAry, gtid);
					break;
				}
			}
			slaveGtids = CollectionUtil.arrayJoin(gtidsAry, ",");
		}
	}
}
